﻿/* Adapted from the Aquiles project: http://aquiles.codeplex.com/. */

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Apache.Cassandra;
using Lucandra.Net.Connection.Endpoint;
using Lucandra.Net.Connection.Pooling;
using System.IO;
using Thrift.Transport;

namespace Lucandra.Net.Connection
{
    internal sealed class CassandraClient : IDisposable
    {
        #region Properties [public]

        public Cassandra.Client InnerClient { get; private set; }
        public CassandraEndpoint Endpoint { get; private set; }
        public string Keyspace { get; private set; }
        public bool IsKeyspaceSet
        {
            get { return this.Keyspace != null; }
        }
        public bool IsOpened
        {
            get
            {
                if ((this.InnerClient != null)
                    && (this.InnerClient.InputProtocol != null)
                    && (this.InnerClient.OutputProtocol != null)
                    && (this.InnerClient.InputProtocol.Transport != null)
                    && (this.InnerClient.OutputProtocol.Transport != null))
                {
                    return this.InnerClient.InputProtocol.Transport.IsOpen && this.InnerClient.OutputProtocol.Transport.IsOpen;
                }
                return false;
            }
        }
        public bool IsBorrowed { get; set; }
        public bool IsInvalidated { get; set; }

        #endregion

        #region Constructors

        public CassandraClient(Cassandra.Client client, CassandraEndpoint endpoint)
        {
            this.InnerClient = client;
            this.Endpoint = endpoint;
            this.Keyspace = null;
        }

        #endregion

        #region Methods [public]

        public void OpenTransport()
        {
            this.InnerClient.InputProtocol.Transport.Open();
            if (!this.InnerClient.InputProtocol.Transport.Equals(this.InnerClient.OutputProtocol.Transport))
                this.InnerClient.OutputProtocol.Transport.Open();
        }

        public void CloseTransport()
        {
            this.InnerClient.InputProtocol.Transport.Close();
            if (!this.InnerClient.InputProtocol.Transport.Equals(this.InnerClient.OutputProtocol.Transport))
                this.InnerClient.OutputProtocol.Transport.Close();
        }

        public void SetKeyspace(string keyspace)
        {
            try
            {
                this.InnerClient.set_keyspace(keyspace);
                this.Keyspace = keyspace;
                return;
            }
            catch (InvalidRequestException ex)
            {
                throw new CassandraException(ex.Why, ex);
            }
            catch (IOException ex)
            {
                CassandraConnectionPool.Invalidate(this);
                throw new CassandraClientInvalidatedException(this, ex);
            }
            catch (TTransportException ex)
            {
                CassandraConnectionPool.Invalidate(this);
                throw new CassandraClientInvalidatedException(this, ex);
            }
        }

        public List<KeySlice> GetRangeSlices(ColumnParent columnParent, SlicePredicate slicePredicate, KeyRange keyRange, ConsistencyLevel consistencyLevel)
        {
            try
            {
                return this.InnerClient.get_range_slices(columnParent, slicePredicate, keyRange, consistencyLevel);
            }
            catch (InvalidRequestException ex)
            {
                throw new CassandraException(ex.Why, ex);
            }
            catch (IOException ex)
            {
                CassandraConnectionPool.Invalidate(this);
                throw new CassandraClientInvalidatedException(this, ex);
            }
            catch (TTransportException ex)
            {
                CassandraConnectionPool.Invalidate(this);
                throw new CassandraClientInvalidatedException(this, ex);
            }
        }

        public void BatchMutate(Dictionary<byte[], Dictionary<string, List<Mutation>>> mutations, ConsistencyLevel consistencyLevel)
        {
            try
            {
                this.InnerClient.batch_mutate(mutations, consistencyLevel);
            }
            catch (InvalidRequestException ex)
            {
                throw new CassandraException(ex.Why, ex);
            }
            catch (IOException ex)
            {
                CassandraConnectionPool.Invalidate(this);
                throw new CassandraClientInvalidatedException(this, ex);
            }
            catch (TTransportException ex)
            {
                CassandraConnectionPool.Invalidate(this);
                throw new CassandraClientInvalidatedException(this, ex);
            }
        }

        #endregion

        public override string ToString()
        {
            return this.Endpoint.ToString();
        }

        ~CassandraClient()
        {
            this.Dispose();
        }

        #region IDisposable Members

        public void Dispose()
        {
            if (this.IsBorrowed)
                CassandraConnectionPool.Release(this);
        }

        #endregion
    }
}
